package com.atbeijing.D08;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 使用KeyedProcessFunction实现sum
 */
public class Example5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env
                .fromElements(0,1,2,3,4,5)
                .keyBy(new KeySelector<Integer, Integer>() {
                    @Override
                    public Integer getKey(Integer value) throws Exception {
                        return value % 2;
                    }
                })
                //分流后每个流有一个处理方法
                .process(new KeyedProcessFunction<Integer, Integer, Integer>() {
                    private ValueState<Integer> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        valueState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("value-state", Types.INT));
                    }

                    @Override
                    public void processElement(Integer value, Context ctx, Collector<Integer> out) throws Exception {
                        //第一次存放
                        if (valueState.value()==null){
                            valueState.update(value);
                        }else {
                            valueState.update(valueState.value()+value);
                        }
                        out.collect(valueState.value());
                    }
                })
                .print();

        env.execute();
    }
}
